package io.kiki.sba.registry.server.data_store.slot;

import io.kiki.sba.registry.common.model.metaserver.CompressPushSwitch;
import io.kiki.sba.registry.common.model.slot.Slot;
import io.kiki.sba.registry.server.shared.remoting.ClientSideExchanger;
import io.kiki.sba.registry.task.TaskErrorSilenceException;
import io.kiki.sba.registry.util.StringFormatter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class SyncLeaderTask implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(CompressPushSwitch.class);

    private final String localDataCenter;
    private final String syncDataCenter;

    private final boolean syncLocalDataCenter;

    private final long startTimestamp = System.currentTimeMillis();
    private final long slotsEpoch;
    private final Slot slot;
    private final SlotDiffSyncer slotDiffSyncer;
    private final ClientSideExchanger clientSideExchanger;
    private final SyncContinues continues;


    public SyncLeaderTask(String localDataCenter, String syncDataCenter, long slotsEpoch, Slot slot, SlotDiffSyncer slotDiffSyncer, ClientSideExchanger clientSideExchanger, SyncContinues continues) {
        this.localDataCenter = localDataCenter;
        this.syncDataCenter = syncDataCenter;

        syncLocalDataCenter = StringUtils.equals(localDataCenter, syncDataCenter);

        this.slotsEpoch = slotsEpoch;
        this.slot = slot;
        this.slotDiffSyncer = slotDiffSyncer;
        this.clientSideExchanger = clientSideExchanger;
        this.continues = continues;


    }

    @Override
    public void run() {
        boolean success = false;
        try {
            success = slotDiffSyncer.syncSlotLeader(localDataCenter, syncDataCenter, syncLocalDataCenter, slot.getId(), slot.getLeaderDataStoreNodeId(), slot.getLeaderEpoch(), clientSideExchanger, slotsEpoch, continues);
            if (!success) {
                throw new RuntimeException(StringFormatter.format("{} sync leader failed", syncDataCenter));
            }
        } catch (Throwable e) {
            logger.error("[syncLeader]syncLocal={}, syncDataCenter={}, failed={}, slot={}", syncDataCenter, syncDataCenter, slot.getLeaderDataStoreNodeId(), slot.getId(), e);
            // rethrow silence exception, notify the task is failed
            throw TaskErrorSilenceException.INSTANCE;
        } finally {
            logger.info("[syncLeader]{},{},{},{},{},span={}", success ? 'Y' : 'N', syncLocalDataCenter ? 'L' : 'R', syncDataCenter, slot.getId(), slot.getLeaderDataStoreNodeId(), System.currentTimeMillis() - startTimestamp);
        }
    }


}
